package four_releaseConfirmation;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;
import util.RabbitMqUtils;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @BelongsProject: RabbitMQ
 * @BelongsPackage: releaseConfirmation
 * @Author: HUI
 * @CreateTime: 2022-04-13 17:00
 * @Description: 发布确认
 */
public class ConfirmationMessage {

    private static final int MESSAGE_COUNT = 1000;


    public static void main(String[] args) throws Exception {
        //1、单个发布确认
//        publishMessageIndividually();//发布1000个单独确认消息，耗时：1739ms

        //2、批量确认
//        publishMessageBatch();//发布1000个批量确认消息，耗时：94ms

        //3、异步发布确认
        publishMessageAsync();//发布1000个异步确认消息，耗时：50ms、发布1000个异步确认消息，耗时：61ms
    }

    public static void publishMessageIndividually() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String uuid = UUID.randomUUID().toString();
        channel.queueDeclare(uuid, true, false, false, null);
        //开启单独发布确认
        channel.confirmSelect();

        long begin = System.currentTimeMillis();
        String message = "";
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            message = i + "";
            channel.basicPublish("", uuid, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            boolean flag = channel.waitForConfirms();
            //发送一次确认一次
            if (flag)
                System.out.println("发送消息完成:" + i);
        }

        long end = System.currentTimeMillis();

        System.err.println("发布" + MESSAGE_COUNT + "个单独确认消息，耗时：" + (end - begin) + "ms");


    }


    public static void publishMessageBatch() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String uuid = UUID.randomUUID().toString();
        channel.queueDeclare(uuid, true, false, false, null);
        //开启单独发布确认
        channel.confirmSelect();

        long begin = System.currentTimeMillis();
        String message = "";
        //批量确认消息大小
        int batchSize = 100;
        //未确认消息个数
        int outstandingMessageCount = 0;
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            message = i + "";
            channel.basicPublish("", uuid, null, message.getBytes());
            outstandingMessageCount++;
            if (outstandingMessageCount == batchSize) {
                channel.waitForConfirms();
                outstandingMessageCount = 0;
            }
        }
        //为了确保还有剩余没有确认消息 再次确认
        if (outstandingMessageCount > 0) {
            channel.waitForConfirms();
        }


        long end = System.currentTimeMillis();

        System.err.println("发布" + MESSAGE_COUNT + "个批量确认消息，耗时：" + (end - begin) + "ms");
    }


    public static void publishMessageAsync() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String uuid = UUID.randomUUID().toString();
        channel.queueDeclare(uuid, true, false, false, null);
        //开启单独发布确认
        channel.confirmSelect();

        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序列号
         * 3.支持并发访问
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

        /**
         * 确认收到消息的一个回调
         * 1.消息序列号
         * 2.true 可以确认小于等于当前序列号的消息
         * false 确认当前序列号消息
         */
        ConfirmCallback ackCallback =(sequenceNumber, multiple) ->{
            if (multiple){
                ConcurrentNavigableMap<Long, String> headMap = outstandingConfirms.headMap(sequenceNumber);
                headMap.clear();
            }else{
                outstandingConfirms.remove(sequenceNumber);
            }
            System.out.println("确认的消息编号："+sequenceNumber);
        };

        /**
         * 未被确认，失败的回调
         */
        ConfirmCallback nackCallback =(sequenceNumber, multiple) ->{
            String s = outstandingConfirms.get(sequenceNumber);
            System.out.println("未被确认的消息编号："+sequenceNumber);
            System.out.println("未被确认的消息"+s);
        };


        /**
         * 添加一个异步确认的监听器
         * 1.确认收到消息的回调
         * 2.未收到消息的回调
         */
        channel.addConfirmListener(ackCallback,nackCallback);

        long begin = System.currentTimeMillis();
        String message = "";
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            message = i + "";
            channel.basicPublish("", uuid, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

            //记录所有要发送的消息，消息的总和
            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);

        }

        long end = System.currentTimeMillis();
        System.err.println("发布" + MESSAGE_COUNT + "个异步确认消息，耗时：" + (end - begin) + "ms");


    }


}
